package com.gitee.cui.concurrency.designmodel.masterworker;

import com.gitee.cui.util.Print;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Master-Worker模式中的Master角色，负责任务的接收、分配、结果汇总等
 * @author cuiqingqiang
 * @version 1.0
 * @date 2021/12/8 15:52
 **/
public class Master<T extends Task, R> {

    /**
     * 所有的worker的集合
     */
    private Map<String, Worker<T, R>> workers = new HashMap();

    /**
     * 任务阻塞队列
     */
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();

    /**
     * 结果集
     */
    private Map<String, R> resultMap = new ConcurrentHashMap<>();

    /**
     * 执行分发任务的线程
     */
    private Thread thread;

    /**
     * 最终的结果之和
     */
    private AtomicInteger sum = new AtomicInteger(0);

    public Master(int workerCount) {
        for (int i = 0; i < workerCount; i++) {
            Worker<T, R> worker = new Worker<>();
            workers.put("子节点：" + i, worker);
        }
        thread = new Thread(this::execute);
        thread.start();
    }

    /**
     * 执行分发任务
     */
    private void execute() {
        while (true) {
            // 取出所有的工作节点，执行任务
            for (Map.Entry<String, Worker<T, R>> entry : workers.entrySet()) {
                try {
                    T task = taskQueue.take();
                    Worker<T, R> worker = entry.getValue();
                    worker.submit(task, this::resultCallBack);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 提交任务
     * @param task 任务
     */
    public void submit(T task) {
        taskQueue.add(task);
    }

    /**
     * 获取最终的结果
     */
    public void printResult() {
        Print.tco("----------sum is :" + sum.get());
        for (Map.Entry<String, R> entry : resultMap.entrySet()) {
            String taskName = entry.getKey();
            Print.fo(taskName + ":" + entry.getValue());
        }
    }

    /**
     * 回调任务
     * @param o 任务
     */
    private void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker:" + task.getWorkerId() + "-Task:" + task.getId();
        R result = task.getResult();
        resultMap.put(taskName, result);
        sum.getAndAdd((Integer) result);
    }
}
